package cn.dmrliu.edu.realtime.app.dws;

import cn.dmrliu.edu.realtime.app.func.DimAsyncFunction;
import cn.dmrliu.edu.realtime.bean.ExamPaperBean;
import cn.dmrliu.edu.realtime.util.ClickhouseUtil;
import cn.dmrliu.edu.realtime.util.DateUtil;
import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

public class DwsToolExamWindow {
    public static void main(String[] args) throws Exception {

        // TODO 1. 基本环境的设置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点的设置

        // TODO 3. 从kafka的主题中读取数据
        String topic = "edu_dwd_tool_exam";
        String group = "edu_dws_tool_exam_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");

        // TODO 4.转为实体类,以 考试id，试卷id，用户id  ， 聚合分数
        SingleOutputStreamOperator<ExamPaperBean> beanDS = kafkaDS.map(new MapFunction<String, ExamPaperBean>() {
            @Override
            public ExamPaperBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                String examId = jsonObject.getString("exam_id");
                String paperId = jsonObject.getString("paper_id");
                String userId = jsonObject.getString("user_id");
                Double score = jsonObject.getDouble("score");
                Long durationSec = jsonObject.getLong("duration_sec");
                String submitTime = jsonObject.getString("submit_time");
                Long ts = DateUtil.toTs(submitTime, true);
                return new ExamPaperBean(
                        "", "",
                        examId, paperId,
                        "", "", "",
                        userId,
                        1L, score, durationSec, null, ts
                );
            }
        });

        SingleOutputStreamOperator<ExamPaperBean> scoreDS = beanDS.keyBy(new KeySelector<ExamPaperBean, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(ExamPaperBean value) throws Exception {
                        return Tuple3.of(value.getExamId(), value.getPaperId(), value.getUserId());
                    }
                })
                .reduce(new ReduceFunction<ExamPaperBean>() {
                    @Override
                    public ExamPaperBean reduce(ExamPaperBean value1, ExamPaperBean value2) throws Exception {
                        value1.setScoreSum(value1.getScoreSum() + value2.getScoreSum());
                        return value1;
                    }
                });

//        scoreDS.print(">>>>>>>");

        // TODO 5. 添加分数段信息

        SingleOutputStreamOperator<ExamPaperBean> assScoreSecDS = scoreDS.map(new MapFunction<ExamPaperBean, ExamPaperBean>() {
            @Override
            public ExamPaperBean map(ExamPaperBean value) throws Exception {
                // 设置分数段加1
                Long[] scoreSection = new Long[6];
                Arrays.fill(scoreSection, 0L);

                Double score = value.getScoreSum();
                if (score.compareTo(100.0) >= 0) {
                    scoreSection[5] = 1L;
                } else if (score.compareTo(80.0) >= 0) {
                    scoreSection[4] = 1L;
                } else if (score.compareTo(60.0) >= 0) {
                    scoreSection[3] = 1L;
                } else if (score.compareTo(40.0) >= 0) {
                    scoreSection[2] = 1L;
                } else if (score.compareTo(20.0) >= 0) {
                    scoreSection[1] = 1L;
                } else {
                    scoreSection[0] = 1L;
                }
                value.setScoreSec(scoreSection);
                return value;
            }
        });


        // TODO 6. 设置水位线
        SingleOutputStreamOperator<ExamPaperBean> watermarkDS = assScoreSecDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ExamPaperBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<ExamPaperBean>() {
                                    @Override
                                    public long extractTimestamp(ExamPaperBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                );

        // TODO 7. 试卷分组，开窗，聚合
        SingleOutputStreamOperator<ExamPaperBean> reduceDS = watermarkDS
                .keyBy(ExamPaperBean::getPaperId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(30)))
                .reduce(new ReduceFunction<ExamPaperBean>() {
                    @Override
                    public ExamPaperBean reduce(ExamPaperBean value1, ExamPaperBean value2) throws Exception {
                        value1.setUserCt(value1.getUserCt() + value2.getUserCt());
                        value1.setScoreSum(value1.getScoreSum() + value2.getScoreSum());
                        Long[] v1ScoreSec = value1.getScoreSec();
                        Long[] v2ScoreSec = value2.getScoreSec();
                        for (int i = 0; i < v1ScoreSec.length; i++) {
                            v1ScoreSec[i] = v1ScoreSec[i] + v2ScoreSec[i];
                        }
                        return value1;
                    }
                }, new WindowFunction<ExamPaperBean, ExamPaperBean, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<ExamPaperBean> input, Collector<ExamPaperBean> out) throws Exception {
                        String sst = DateUtil.toYmdHms(window.getStart());
                        String edt = DateUtil.toYmdHms(window.getEnd());
                        for (ExamPaperBean examPaperBean : input) {
                            examPaperBean.setStt(sst);
                            examPaperBean.setEdt(edt);
                            examPaperBean.setTs(System.currentTimeMillis());
                            // 小数位保留2为
                            String two = String.format("%.2f", examPaperBean.getScoreSum());
                            examPaperBean.setScoreSum(Double.parseDouble(two));
                            out.collect(examPaperBean);
                        }
                    }
                });
//        reduceDS.print(">>>>>>>>>>");

        // TODO 8. 补充维度信息
        SingleOutputStreamOperator<ExamPaperBean> withPaperDimDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ExamPaperBean>("dim_test_paper") {
                    @Override
                    public String getKey(ExamPaperBean examPaperBean) {
                        return examPaperBean.getPaperId();
                    }

                    @Override
                    public void join(ExamPaperBean examPaperBean, JSONObject dimInfo) {
                        examPaperBean.setPageName(dimInfo.getString("paper_title".toUpperCase()));
                        examPaperBean.setCourseId(dimInfo.getString("course_id".toUpperCase()));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<ExamPaperBean> withCourseDimDS = AsyncDataStream.unorderedWait(
                withPaperDimDS,
                new DimAsyncFunction<ExamPaperBean>("dim_course_info") {
                    @Override
                    public String getKey(ExamPaperBean examPaperBean) {
                        return examPaperBean.getCourseId();
                    }

                    @Override
                    public void join(ExamPaperBean examPaperBean, JSONObject dimInfo) {
                        examPaperBean.setCourseName(dimInfo.getString("course_name".toUpperCase()));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        withCourseDimDS.print("》》》》》》》》》");

        // TODO 9. 写入clickhouse中
        withCourseDimDS.addSink(ClickhouseUtil.getSinkFunction("insert into dws_tool_exam_page_window values(?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
